package org.zero.example.springbatchdemo.config;

import lombok.extern.slf4j.Slf4j;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
import org.springframework.batch.core.launch.support.RunIdIncrementer;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.item.database.JpaPagingItemReader;
import org.springframework.batch.item.database.orm.JpaNativeQueryProvider;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.zero.example.springbatchdemo.model.Student;
import org.zero.example.springbatchdemo.task.listener.JobListener;

import javax.persistence.EntityManagerFactory;

/**
 * 配置一个最基本的Job
 * 一个Job 通常由一个或多个Step组成(基本就像是一个工作流)；
 * 一个Step通常由三部分组成(读入数据 ItemReader，处理数据 ItemProcessor，写入数据 ItemWriter)
 * :@EnableBatchProcessing提供用于构建批处理作业的基本配置
 *
 * @author 01
 * @date 2019-02-24
 **/
@Slf4j
@Configuration
@EnableBatchProcessing
public class DataBatchConfiguration {
    /**
     * Job构建工厂，用于构建Job
     */
    private final JobBuilderFactory jobBuilderFactory;

    /**
     * Step构建工厂，用于构建Step
     */
    private final StepBuilderFactory stepBuilderFactory;

    /**
     * 实体管理工工厂，用于访问数据
     */
    private final EntityManagerFactory emf;

    /**
     * 自定义的简单Job监听器
     */
    private final JobListener jobListener;

    @Autowired
    public DataBatchConfiguration(JobBuilderFactory jobBuilderFactory, StepBuilderFactory stepBuilderFactory,
                                  EntityManagerFactory emf, JobListener jobListener) {
        this.jobBuilderFactory = jobBuilderFactory;
        this.stepBuilderFactory = stepBuilderFactory;
        this.emf = emf;
        this.jobListener = jobListener;
    }

    /**
     * 一个简单基础的Job通常由一个或者多个Step组成
     */
    @Bean
    public Job dataHandleJob() {
        return jobBuilderFactory.get("dataHandleJob").
                incrementer(new RunIdIncrementer()).
                // start是JOB执行的第一个step
                        start(handleDataStep()).
                // next(xxxStep()).
                // next(xxxStep()).
                // ...
                // 设置了一个简单JobListener
                        listener(jobListener).
                        build();
    }

    /**
     * 一个简单基础的Step主要分为三个部分
     * ItemReader : 用于读取数据
     * ItemProcessor : 用于处理数据
     * ItemWriter : 用于写数据
     */
    @Bean
    public Step handleDataStep() {
        return stepBuilderFactory.get("getData").
                // <输入,输出> 。chunk通俗的讲类似于SQL的commit; 这里表示处理(processor)100条后写入(writer)一次
                        <Student, Student>chunk(100).
                // 捕捉到异常就重试,重试100次还是异常,JOB就停止并标志失败
                        faultTolerant().retryLimit(3).retry(Exception.class).skipLimit(100).skip(Exception.class).
                // 指定ItemReader
                        reader(getDataReader()).
                // 指定ItemProcessor
                        processor(getDataProcessor()).
                // 指定ItemWriter
                        writer(getDataWriter()).
                        build();
    }

    @Bean
    public ItemReader<? extends Student> getDataReader() {
        // 读取数据,这里可以用JPA,JDBC,JMS 等方式 读入数据
        JpaPagingItemReader<Student> reader = new JpaPagingItemReader<>();
        // 这里选择JPA方式读数据 一个简单的 native SQL
        String sqlQuery = "SELECT * FROM student";

        try {
            JpaNativeQueryProvider<Student> queryProvider = new JpaNativeQueryProvider<>();
            queryProvider.setSqlQuery(sqlQuery);
            queryProvider.setEntityClass(Student.class);
            queryProvider.afterPropertiesSet();
            reader.setEntityManagerFactory(emf);
            reader.setPageSize(3);
            reader.setQueryProvider(queryProvider);
            reader.afterPropertiesSet();

            // 所有ItemReader和ItemWriter实现都会在ExecutionContext提交之前将其当前状态存储在其中,
            // 如果不希望这样做,可以设置setSaveState(false)
            reader.setSaveState(true);
        } catch (Exception e) {
            e.printStackTrace();
        }

        return reader;
    }

    @Bean
    public ItemProcessor<Student, Student> getDataProcessor() {
        return student -> {
            //模拟处理数据，这里处理就是打印一下
            log.info("processor data : " + student.toString());

            return student;
        };
    }

    @Bean
    public ItemWriter<Student> getDataWriter() {
        return list -> {
            for (Student student : list) {
                //模拟写数据，这里写真正写入数据的逻辑
                log.info("write data : " + student);
            }
        };
    }
}
